package org.dragonnova.meetingserver.core.manager;

import android.annotation.SuppressLint;
import android.content.Context;
import android.content.Intent;
import android.net.wifi.WifiManager;
import android.os.Handler;
import android.os.Message;
import android.support.annotation.NonNull;
import android.text.TextUtils;
import android.util.Log;
import android.widget.Toast;

import com.iflytek.XFSynthesizer;
import com.ztspeech.AppAudio;
import com.ztspeech.ZTTranslate;

import org.dragonnova.meetingserver.R;
import org.dragonnova.meetingserver.core.SocketUtils;
import org.dragonnova.meetingserver.core.net.socket.NetWorker;
import org.dragonnova.meetingserver.core.net.socket.UdpPacket;
import org.dragonnova.meetingserver.core.net.socket.UdpSocket;
import org.dragonnova.meetingserver.model.AObservable;
import org.dragonnova.meetingserver.model.IObserver;
import org.dragonnova.meetingserver.model.bean.DeviceInfo;
import org.dragonnova.meetingserver.model.bean.MeetingInfo;
import org.dragonnova.meetingserver.model.bean.VoiceInfo;
import org.dragonnova.meetingserver.server.application.TransBoxApplication;
import org.dragonnova.meetingserver.server.service.TranslationService;
import org.dragonnova.meetingserver.server.service.WIFIConnectionFactory;
import org.dragonnova.meetingserver.utils.Constants;
import org.dragonnova.meetingserver.utils.ContentResolverUtil;
import org.dragonnova.meetingserver.utils.LogUtil;
import org.dragonnova.meetingserver.utils.PromptUtil;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by hucn on 2016/6/12.
 * Description:
 */
public class NettyManager implements NetManager {

    private final static String TAG = NettyManager.class.getName();
    /**
     * action
     */
    private static final byte KEY_ACTION_GROUP_NO_JOIN = (byte) 0xBB;
    private static final byte KEY_ACTION_GROUP_ALLOW_JOIN = (byte) 0xAA;
    private static final byte KEY_ACTION_GROUP_JOINED = (byte) 0xDD;
    private static final int KEY_ACTION_GROUP_CONFIRM_1 = 0xEE;
    private static final int KEY_ACTION_GROUP_JOIN_REQUEST = 0xCC;
    private static final int KEY_ACTION_GROUP_CONFIRM_2 = 0xFF;
    private static final int KEY_ACTION_GROUP_CONNECTED_REQUEST = 0x01;
    private static final int KEY_ACTION_GROUP_CONNECTED_RESPONSE = 0x02;
    private static final int KEY_ACTION_GROUP_EXIT = 0x99;
    private static final int KEY_ACTION_GROUP_CHANGE_LANGUAGE = 0x31;
    private static final int KEY_ACTION_GROUP_CHANGE_LANGUAGE_COMPLETE = 0x32;
    private static final int KEY_ACTION_GROUP_RECEIVE_ERROR = 0x33;
    private static final int KEY_ACTION_VOICE_REQUEST_START = 0x03;
    private static final int KEY_ACTION_VOICE_REQUEST_STOP = 0x04;
    private static final int KEY_ACTION_VOICE_REQUEST_PERSIST = 0x08;
    private static final int KEY_ACTION_VOICE_RESPONSE_PERMIT = 0x05;
    private static final int KEY_ACTION_VOICE_RESPONSE_PREVENT = 0x06;
    private static final int KEY_ACTION_VOICE_RESPONSE_FINISH = 0x07;
    private static final int KEY_ACTION_VOICE_RESPONSE_PERSIST = 0x09;
    private static final int KEY_ACTION_TRANSFER_VOICE_DATA = 0x10;
    private static final int KEY_ACTION_CREATE_MEETING_REQUEST = 0x19;
    private static final int KEY_ACTION_CREATE_MEETING_RESPONSE_SUCCESS = 0x18;
    private static final int KEY_ACTION_CREATE_MEETING_RESPONSE_ALREADY_CREATED = 0x17;
    private static final int KEY_ACTION_REQUEST_MEETING_REQUEST = 0x16;
    private static final int KEY_ACTION_REQUEST_MEETING_RESPONSE_HELDING = 0x15;
    private static final int KEY_ACTION_MEETING_RESPONSE_OVER = 0x14;
    private static final int KEY_ACTION_CLOSE_MEETING_REQUEST = 0x13;
    private static final int KEY_ACTION_CLOSE_MEETING_RESPONSE_FINISH = 0x12;
    private static final int KEY_ACTION_CLOSE_MEETING_RESPONSE_HAS_OVER = 0x11;

    /**
     * 多播地址
     */
    // private final static String MULTICAST_NETWORK_IP = "239.255.255.250";
    private final static String MULTICAST_NETWORK_IP = "224.0.0.1";
    /**
     * 多播端口
     */
    private final static int MULTICAST_NETWORK_PORT = 25221;
    /**
     * 队列上限
     */
    private final static int VOICE_MAX_QUEUE_SIZE = 200;
    /**
     * 翻译请求次数
     */
    private final static int MAX_TASK_COUNT = 3;
    /**
     * UDP通信端口
     */
    private final static int SERVER_SOCKET_PORT = 30002;
    /**
     * UDP通信端口
     */
    private final static int SERVER_SOCKET_PORT_BEAT = 30001;
    /**
     * UDP通信端口
     */
    private final static int SERVER_SOCKET_PORT_VOICE_REQUEST = 30000;
    /**
     * UDP服务端IP
     */
    public static String UDP_SERVER_IP;
    /**
     * 保存语音的队列，用于服务端的翻译
     */
    private LinkedBlockingQueue<VoiceInfo> voiceInfoTransQueue = new LinkedBlockingQueue<>(VOICE_MAX_QUEUE_SIZE);
    /**
     * 保存语音的队列，用于传输给其他连接的客户端
     */
    private LinkedBlockingQueue<VoiceInfo> voiceInfoBroadcastQueue = new LinkedBlockingQueue<>(VOICE_MAX_QUEUE_SIZE);
    /**
     * 用于注册观察者
     */
    private WIFIConnectionFactory factory = WIFIConnectionFactory.getInstance();
    /**
     * 翻译任务次数
     */
    private AtomicInteger task_count = new AtomicInteger(0);
    /**
     * 用于UDP多播组网
     */
    private UdpSocket mMulticastSocket;
    /**
     * 用于管理WIFI网络入口
     */
    private MainThread mainWorker;
    /**
     * 组网接收线程
     */
    private MulticastRecThread multicastRecWorker;
    /**
     * UDP接收线程
     */
    private UdpWorker udpWorker;
    /**
     * UDP语音接收和发送线程
     */
    public BeatWorker beatWorker;
    /**
     * UDP语音接收和发送线程
     */
    public VoiceResWorker voiceResWorker;
    /**
     * 多播锁
     */
    private WifiManager.MulticastLock multicastLock;
    // 翻译线程池
    private ExecutorService producerExecutorService;
    // 语音包越位
    private VoiceProducer mProducer;
    private VoiceConsumer mConsumer;
    // 关于翻译的语音类
    private XFSynthesizer xfSynthesizer;//合成类
    // 会议信息
    private MeetingManager mMeetingManager;
    // Context
    private Context mContext;
    // wifiManager
    private WifiManager mWifiManager;
    // socket是否已经启动
    private boolean isStart = false;
    //连接的设备管理
    private DeviceManager mDeviceManger;

    public NettyManager(Context context, WifiManager wifiManager) {
        mContext = context;
        mWifiManager = wifiManager;
        factory.getObservable().addObserver(new WIFIObserver());
        //翻译对象
        xfSynthesizer = new XFSynthesizer(mHandler, mContext);
        // 设备管理
        mDeviceManger = DeviceManager.getInstance(mContext.getApplicationContext());
        // 会议管理
        mMeetingManager = MeetingManager.getInstance(mContext);
        init();
    }

    /**
     * 处理事件的handler
     */
    @SuppressLint("HandlerLeak")
    private Handler mHandler = new Handler() {
        Intent intent;

        public void handleMessage(Message msg) {
            switch (msg.what) {
                case AppAudio.TRANSLATE_ERROR:
                    Toast.makeText(mContext, "翻译错误", Toast.LENGTH_SHORT).show();
                    break;
                case AppAudio.TRANSLATED:
                    if (msg.arg1 == -1)
                        Toast.makeText(mContext, "录音时间请小于60s", Toast.LENGTH_SHORT).show();
                    else if (msg.arg1 == 1 && msg.arg2 == 1) {
                        VoiceInfo voiceInfo = (VoiceInfo) msg.obj;
                        voiceInfo.setFlag(1);
                        voiceInfo.setOrgCostTime(1);
                        String userId = TransBoxApplication.getInstance().getCurrentUserName();
                        if (TextUtils.isEmpty(userId)) {
                            userId = "null";
                        }
                        voiceInfo.setUserId(userId);
                        //在数据库中保存聊天记录
                        ContentResolverUtil resolverUtil = new ContentResolverUtil(mContext);
                        int id = resolverUtil.insertChatHistory(voiceInfo);
                        //刷新页面
                        intent = new Intent();
                        intent.setAction(Constants.TRANS_BORADCAST_ACTION);
                        intent.putExtra(Constants.BORADCAST_INTENT_OPTION, TranslationService.EXTRA_TRANSLATED);
                        intent.putExtra("voiceInfo", voiceInfo);
                        mContext.sendBroadcast(intent);
                        if (voiceInfo.getType() == 'c') {
                            xfSynthesizer.start(voiceInfo.getTranContent(), true);
                        } else {
                            xfSynthesizer.start(voiceInfo.getTranContent(), false);
                        }
                    }
                    break;
            }
        }
    };


    public void startMain() {
        //启动网络线程，组网、服务SOCKET初始化
        mainWorker = new MainThread();
        mainWorker.start();
        synchronized (mainWorker) {
            mainWorker.notify();
        }
    }

    public void stopMain() {
        if (mainWorker != null) {
            mainWorker.close();
            mainWorker.setFlag(false);
            mainWorker.interrupt();
        }
    }

    private void startMulti() {
        InetAddress multiInetAddr = null;
        MulticastSocket multicastSocket = null;

        if (multicastSocket == null || !multicastSocket.isBound()) {
            try {
                // 加锁
                multicastLock();
                //启动多播套接字
                multicastSocket = new MulticastSocket(MULTICAST_NETWORK_PORT);
                multiInetAddr = InetAddress.getByName(MULTICAST_NETWORK_IP);
                multicastSocket.joinGroup(multiInetAddr);
                multicastSocket.setReuseAddress(true);
                // 设置本MulticastSocket发送的数据报被回送到自身
                multicastSocket.setLoopbackMode(true);
                // 该ttl参数设置数据报最多可以跨过多少个网络，当ttl为0时，指定数据报应停留在本地主机；当ttl的值为1时，指定数据报发送到本地局域网；当ttl的值为32时，意味着只能发送到本站点的网络上；当ttl为64时，意味着数据报应保留在本地区；当ttl的值为128时，意味着数据报应保留在本大洲；当ttl为255时，意味着数据报可发送到所有地方；默认情况下，该ttl的值为1。
                multicastSocket.setTimeToLive(1);
                // 封装
                mMulticastSocket = new UdpSocket(multicastSocket);
                Log.e(TAG, "MulticastSocket Join Group ");
            } catch (IOException e) {
                e.printStackTrace();
                if (multicastSocket != null) {
                    try {
                        multicastSocket.leaveGroup(multiInetAddr);
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    } finally {
                        multicastSocket.close();
                    }
                }
                multicastUnLock();
            }
            if (isMulticastSocketExisted()) {
                try {
                    // 启动组网接收线程
                    multicastRecWorker = new MulticastRecThread(mMulticastSocket);
                    multicastRecWorker.start();
                    Log.e(TAG, "MulticastRecWorker Thread Start.");
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void startNetWorkers() {
        //监听组网情况的线程
        Log.e(TAG, "NetWorker Thread Start. ");
        int count = 0;
        DatagramSocket beatSocket = null;
        try {
            beatSocket = new DatagramSocket(new InetSocketAddress(
                    UDP_SERVER_IP, SERVER_SOCKET_PORT_BEAT));
            beatSocket.setSoTimeout(0);
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (beatSocket != null) {
            UdpSocket beatSocket1 = new UdpSocket(beatSocket);
            beatWorker = new BeatWorker(beatSocket1);
            beatWorker.start();
            count++;
            LogUtil.e(TAG, "BeatWorker Thread Start. ");
        }

        //应答机制的线程
        DatagramSocket voiceResSocket = null;
        try {
            voiceResSocket = new DatagramSocket(new InetSocketAddress(
                    UDP_SERVER_IP, SERVER_SOCKET_PORT_VOICE_REQUEST));
            voiceResSocket.setSoTimeout(1500);
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (voiceResSocket != null) {
            UdpSocket voiceReqSocket1 = new UdpSocket(voiceResSocket);
            voiceResWorker = new VoiceResWorker(voiceReqSocket1);
            voiceResWorker.start();
            count++;
            LogUtil.e(TAG, "VoiceResWorker Thread Start. ");
        }

        //启动UDP语音包接收线程
        DatagramSocket udpSocket = null;
        try {
            udpSocket = new DatagramSocket(new InetSocketAddress(
                    UDP_SERVER_IP, SERVER_SOCKET_PORT));
            udpSocket.setSoTimeout(0);
        } catch (SocketException e) {
            e.printStackTrace();
        }
        if (udpSocket != null) {
            UdpSocket udpSocket1 = new UdpSocket(udpSocket);
            //客户端组播包 由ID和IP组成
            udpWorker = new UdpWorker(udpSocket1);
            udpWorker.start();
            count++;
            LogUtil.e(TAG, "UdpWorker Thread Start. ");
        }
        if (count != 3) {
            LogUtil.e(TAG, "netWorker Thread Start at Count = " + count);
        }
    }

    public void startAllThreadAndSocket() {
        startMain();
        startMulti();
        startOtherThread();
        startNetWorkers();
    }

    private void startOtherThread() {
        // 发送数据到其他客户端的线程
        mConsumer = new VoiceConsumer();
        mConsumer.start();
        // 处理接收的udp包的线程
        mProducer = new VoiceProducer();
        mProducer.start();
    }

    private void stopMulti() {
        if (multicastRecWorker != null) {
            multicastRecWorker.setFlag(false);
            multicastRecWorker.close();
            multicastRecWorker = null;
        }
        if (mMulticastSocket != null) {
            mMulticastSocket.close();
            mMulticastSocket = null;
        }
    }

    private void stopNetWorkers() {
        if (udpWorker != null) {
            udpWorker.setFlag(false);
            udpWorker.close();
            udpWorker = null;
        }
        if (voiceResWorker != null) {
            voiceResWorker.setFlag(false);
            voiceResWorker.close();
            voiceResWorker = null;
        }

        if (beatWorker != null) {
            beatWorker.setFlag(false);
            beatWorker.close();
            beatWorker = null;
        }
    }

    private void stopOtherThread() {
        if (mProducer != null) {
            mProducer.interrupt();
            mProducer = null;
        }
        if (mConsumer != null) {
            mConsumer.setFlag(false);
            mConsumer.interrupt();
            mConsumer = null;
        }
    }

    /**
     * 销毁除了主线程和网络监听线程外的其他线程
     */
    private void stopAllThreadAndSocket() {
        stopMain();
        stopMulti();
        stopOtherThread();
        stopNetWorkers();
    }

    /**
     * 更新状态，通过主线程
     */
    public void refreshClientDevice(DeviceInfo deviceInfo) {
        if (mDeviceManger.contains(deviceInfo.getHostId())) {
            mDeviceManger.update(deviceInfo);
        } else {
            mDeviceManger.add(deviceInfo);
        }
    }

    public boolean containsDeviceInfo(DeviceInfo info) {
        return mDeviceManger.contains(info);
    }

    /**
     * 查找设备信息
     */
    public DeviceInfo findDeviceInfo(String clientId) {
        return mDeviceManger.getDeviceInfo(clientId);
    }

    public boolean removeDeviceInfo(DeviceInfo deviceInfo) {
        return mDeviceManger.remove(deviceInfo);
    }

    /**
     * 发送数据至客户端
     */
    public void sendPacketData(char language, SocketAddress addr, byte[] data) throws IOException {
        sendPacketData(language, addr, data, 0, data.length);
    }

    /**
     * 发送语音包向指定的地址
     */
    public synchronized void sendPacketData(char language, SocketAddress addr, byte[] data, int offset, int length)
            throws IOException {
        List<DeviceInfo> deviceInfoList = mDeviceManger.getConnettedDeviceInfos();
        for (DeviceInfo info : deviceInfoList) {
            if (info.getWifiAddr() != null) {
                SocketAddress addr1 = new InetSocketAddress(info.getWifiAddr(), info.getPort());
                udpWorker.sendPacketData(data, new byte[]{KEY_ACTION_TRANSFER_VOICE_DATA}, addr1);
            }
        }
    }

    public synchronized void sendPacketData(byte[] data, byte[] action) throws IOException {
        //根据客户端列表返回信息
        List<DeviceInfo> deviceInfoList = mDeviceManger.getConnettedDeviceInfos();
        for (DeviceInfo info : deviceInfoList) {
            if (info.getWifiAddr() != null) {
                SocketAddress addr1 = new InetSocketAddress(info.getWifiAddr(), info.getPort());
                udpWorker.sendPacketData(data, action, addr1);
            }
        }
    }

    public boolean isMulticastSocketExisted() {
        return mMulticastSocket != null;
    }

    /**
     * 加锁
     */
    private synchronized void multicastLock() {
        multicastLock = mWifiManager.createMulticastLock("GROUP LOCK");
        multicastLock.acquire();
    }

    /**
     * 解锁
     */
    private synchronized void multicastUnLock() {
        if (multicastLock != null && multicastLock.isHeld())
            multicastLock.release();
    }

    /**
     * 关闭线程池
     */
    public void shutDownThreadPool() {
        if (producerExecutorService != null) {
            producerExecutorService.shutdown();
            try {
                while (!producerExecutorService.isTerminated())
                    producerExecutorService.awaitTermination(5000, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        producerExecutorService = null;
    }

    @Override
    public void init() {
        int poolSize = Runtime.getRuntime().availableProcessors() * 2;
        producerExecutorService = Executors.newFixedThreadPool(poolSize);
    }

    @Override
    public void destory() {
        shutDownThreadPool();
    }

    @Override
    public void start() {
        //启动网络线程，组网、服务SOCKET初始化
        startAllThreadAndSocket();
        isStart = true;
    }

    @Override
    public void stop() {
        multicastUnLock();
        stopAllThreadAndSocket();
        isStart = false;
    }

    @Override
    public void enableNetWorker() {
        /*synchronized (mainWorker) {
            mainWorker.notify();
        }*/
    }

    @Override
    public void disableNetWorker() {
        stopNetWorkers();
    }

    @Override
    public void tryGroup() {

    }

    @Override
    public void tryDisGroup() {

    }

    @Override
    public void setServerIp(String udpServerIp) {
        UDP_SERVER_IP = udpServerIp;
    }

    @Override
    public boolean isStarted() {
        return isStart;
    }

    /**
     * 主线程，初始化WIFI连接，组播，服务端SOCKET
     */
    private class MainThread extends Thread {
        private boolean waitFlag = true;

        @Override
        public void run() {
            while (waitFlag) {
                try {
                    synchronized (this) {
                        wait();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return;
                }
            }

        }

        public void close() {
            multicastUnLock();
        }

        public void setFlag(boolean flag) {
            this.waitFlag = flag;
        }
    }

    /**
     * 组网接收线程
     */
    private class MulticastRecThread extends NetWorker {

        public MulticastRecThread(UdpSocket udpSocket) {
            super(udpSocket);
        }

        @Override
        public void work() {
            while (waitFlag) {
                try {
                    // 客户端组播包 由ID和IP组成
                    UdpPacket packet = mMulticastSocket.receiveMsgFromNetwork(UdpSocket.PACKET_BUFFER_SIZE);
                    DatagramPacket inPacket = packet.getPacket();
                    if (inPacket != null && inPacket.getLength() > 0) {
                        Log.i(TAG, " receive Data from Remote: "
                                + inPacket.getAddress().getHostAddress());
                        doHandlerInPacket(packet);
                    } else {
                        Log.i(TAG, " receive Data Is Null");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        private void doHandlerInPacket(UdpPacket packet) {
            int type = SocketUtils.byte2Int(packet.getMessage().getAction());
            DatagramPacket inPacket = packet.getPacket();
            DeviceInfo deviceInfo;
            switch (type) {
                case KEY_ACTION_GROUP_JOIN_REQUEST:
                    deviceInfo = DeviceInfo.parse(new String(packet.getMessage().getPacketData()));
                    boolean exist = containsDeviceInfo(deviceInfo);
                    // 存在
                    try {
                        if (exist) {
                            sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_ALLOW_JOIN}, inPacket.getSocketAddress());
                        } else {
                            sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_NO_JOIN}, inPacket.getSocketAddress());
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    break;
                case KEY_ACTION_GROUP_CONFIRM_1:
                    try {
                        sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_ALLOW_JOIN}, inPacket.getSocketAddress());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    break;
                case KEY_ACTION_GROUP_CONFIRM_2:
                    deviceInfo = DeviceInfo.parse(new String(packet.getMessage().getPacketData()));
                    if (deviceInfo == null) break;
                    // 加入
                    deviceInfo.setWifiStatus(DeviceInfo.Status.Connected);
                    refreshClientDevice(deviceInfo);
                    // 存在
                    try {
                        sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_JOINED}, inPacket.getSocketAddress());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    PromptUtil.poolPlay(mContext, R.raw.client_grouped);
                    break;
                case KEY_ACTION_GROUP_EXIT:
                    deviceInfo = DeviceInfo.parse(new String(packet.getMessage().getPacketData()));
                    removeDeviceInfo(deviceInfo);
                    break;
                case KEY_ACTION_GROUP_CHANGE_LANGUAGE:
                    deviceInfo = DeviceInfo.parse(new String(packet.getMessage().getPacketData()));
                    if (deviceInfo == null) break;
                    deviceInfo.setWifiStatus(DeviceInfo.Status.Connected);
                    refreshClientDevice(deviceInfo);
                    // 通知客户端
                    try {
                        sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_CHANGE_LANGUAGE_COMPLETE}, inPacket.getSocketAddress());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    break;
                default:
                    //PromptUtil.poolPlay(mContext, R.raw.group_content_error);
                    try {
                        sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_RECEIVE_ERROR}, inPacket.getSocketAddress());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    break;
            }
        }
    }

    /**
     * 用于维护客户端数据传输
     */
    public class UdpWorker extends NetWorker {

        public UdpWorker(UdpSocket udpSocket) {
            super(udpSocket);
        }

        @Override
        public void work() {
            while (waitFlag && isConnected()) {
                try {
                    UdpPacket packet = receiveMessage(UdpSocket.PACKET_BUFFER_SIZE);
                    if (packet.hasData()) {
                        LogUtil.i(TAG, " receive Data from Remote: "
                                + ((InetSocketAddress) packet.getSocketAddress()).getHostName());
                        int action = SocketUtils.byte2Int(packet.getMessage().getAction());
                        switch (action) {
                            case KEY_ACTION_TRANSFER_VOICE_DATA:
                                factory.getObservable().handler(packet);
                                break;
                            case KEY_ACTION_CREATE_MEETING_REQUEST:
                                MeetingInfo meetingInfo = MeetingInfo.parse(new String((packet.getMessage().getPacketData())));
                                LogUtil.d("create meeting:" + meetingInfo.toString());
                                if (mMeetingManager.isMeetingCreated()) {
                                    sendPacketData(mMeetingManager.getcurrentMeeting().encode(), new byte[]{KEY_ACTION_CREATE_MEETING_RESPONSE_ALREADY_CREATED}, packet.getSocketAddress());
                                } else {
                                    meetingInfo.setFlag(1);
                                    mMeetingManager.createMeeting(meetingInfo);
                                    sendPacketData(mMeetingManager.getcurrentMeeting().encode(), new byte[]{KEY_ACTION_CREATE_MEETING_RESPONSE_SUCCESS}, packet.getSocketAddress());
                                }
                                break;
                            case KEY_ACTION_REQUEST_MEETING_REQUEST:
                                LogUtil.d("request meeting:" + mMeetingManager.getcurrentMeeting());
                                if (mMeetingManager.isMeetingCreated()) {
                                    sendPacketData(mMeetingManager.getcurrentMeeting().encode(), new byte[]{KEY_ACTION_REQUEST_MEETING_RESPONSE_HELDING}, packet.getSocketAddress());
                                } else {
                                    sendPacketData(new byte[0], new byte[]{KEY_ACTION_MEETING_RESPONSE_OVER}, packet.getSocketAddress());
                                }
                                break;
                            case KEY_ACTION_CLOSE_MEETING_REQUEST:
                                LogUtil.d("closing meeting:" + mMeetingManager.getcurrentMeeting());
                                if (mMeetingManager.isMeetingCreated()) {
                                    mMeetingManager.closeMeeting();
                                    NettyManager.this.sendPacketData(new byte[0], new byte[]{KEY_ACTION_CLOSE_MEETING_RESPONSE_FINISH});
                                } else {
                                    sendPacketData(new byte[0], new byte[]{KEY_ACTION_CLOSE_MEETING_RESPONSE_HAS_OVER}, packet.getSocketAddress());
                                }
                                break;
                            default:
                                //PromptUtil.poolPlay(mContext, R.raw.group_content_error);
                                break;
                        }
                    } else
                        LogUtil.i(TAG, " receive Data Is Null");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * WIFI观察者
     */
    public class WIFIObserver implements IObserver {
        @Override
        public void doWhat(AObservable object, Object... args) {
            //播放语音
            UdpPacket packet = (UdpPacket) args[0];
            String msg = new String(packet.getMessage().getPacketData());
            VoiceInfo voiceInfo = VoiceInfo.parse(msg);
            LogUtil.eee(msg);

            if (voiceInfo == null) {
                // TODO: 2015/12/25 未找到语音发过来的设备地址
                //PromptUtil.poolPlay(mContext, R.raw.group_content_error);
                return;
            }
            // 加入队列
            voiceInfoTransQueue.add(voiceInfo);
        }
    }

    /**
     * 对传入的语音数据进行处理
     */
    private class VoiceProducer extends Thread {

        boolean mFlag = true;

        final ZTTranslate translate = new ZTTranslate(NettyManager.this.mHandler);

        private volatile VoiceInfo voiceInfo = null;

        private Runnable transfer_runnable = new Runnable() {

            @Override
            public void run() {
                //翻译请求
                doTransfer(voiceInfo);
            }
        };

        public VoiceProducer() {
            setName(VoiceProducer.class.getName());
        }

        /**
         * 云端识别、翻译、合成
         */
        synchronized void doTransfer(@NonNull VoiceInfo data) throws IndexOutOfBoundsException {
            task_count.set(0);
            if (data.getType() == 'c') {
                translate.setLanguage(ZTTranslate.LANGUAGE_MANDARIN);
            } else if (data.getType() == 'e') {
                translate.setLanguage(ZTTranslate.LANGUAGE_ENGLISH);
            }
            try {
                //请求翻译
                String transferStr = translate.start(data.getOrgContent());
                if (transferStr != null) transferStr = transferStr.trim();
                data.setTranContent(transferStr);
                //放入队列
                voiceInfoBroadcastQueue.add(voiceInfo);
                task_count.set(task_count.incrementAndGet());
            } catch (Exception e) {
                if (task_count.intValue() < MAX_TASK_COUNT) {
                    doTransfer(data);
                } else {
                    throw new IndexOutOfBoundsException("Transfer Commit Over " + MAX_TASK_COUNT + " Times");
                }
            }
        }

        @Override
        public void run() {
            try {
                while (mFlag || !Thread.interrupted()) {
                    voiceInfo = voiceInfoTransQueue.take();
                    if (producerExecutorService == null) {
                        LogUtil.eee("producerExecutorService == null");
                    }
                    if (transfer_runnable == null) {
                        LogUtil.eee("transfer_runnable == null");
                    }
                    producerExecutorService.execute(transfer_runnable);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 将语音处理结果返回客户端
     */
    private class VoiceConsumer extends Thread {

        boolean mFlag = true;

        @Override
        public void run() {
            try {
                while (mFlag || !Thread.interrupted()) {
                    //返回客户端
                    VoiceInfo voiceInfo = voiceInfoBroadcastQueue.take();
                    try {
                            /*VoiceInfo{id=0 orgData=null tranData=null orgContent='I saw you in.' tranContent='我看到你在。' inserttime=1460710583530 type=e orgPath='null' orgAddress=null orgCosttime=0 userId='null' flag=0}*/
                        //VoiceInfo voiceInfo = voiceInfoTransQueue.remove();
                        // TODO: 2016/1/8 返回客户端翻译的语
                        byte[] result = voiceInfo.encode();
                        if (result.length == 0) continue;
                        //MIPS 转换大端
                        Log.i(TAG, "接收到并转发数据--" + voiceInfo.toString() + "--");
                        NettyManager.this.sendPacketData(voiceInfo.getType(), voiceInfo.getOrgAddress(), result);
                        //通知Handler
                        mHandler.obtainMessage(AppAudio.TRANSLATED, 1, 1, voiceInfo).sendToTarget();
                        Log.d(TAG, "ExecutorService Transfer Data Id-" + voiceInfo.getId() + " Complete"
                                + "SendData Successful When " + System.currentTimeMillis());

                    } catch (NullPointerException e) {
                        Log.e(TAG, "VoiceInfo is Null " + e.getMessage());
                    } catch (NoSuchElementException | IOException e) {
                        Log.e(TAG, "", e);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        /**
         * 结束此线程
         */
        public void setFlag(boolean flag) {
            this.mFlag = flag;
        }
    }

    /**
     * 用于检查网络状况
     */
    class BeatWorker extends NetWorker {

        private Timer timer;

        public BeatWorker(UdpSocket udpSocket) {
            super(udpSocket);
        }

        @Override
        public void work() {
            timer = new Timer();
            timer.schedule(new TimerTask() {
                public void run() {
                    List<DeviceInfo> infoList = mDeviceManger.getAllDeviceInfos();
                    for (DeviceInfo info : infoList) {
                        /*if (info.getWifiStatus() == DeviceInfo.Status.Connected) {
                            info.setDisconnectedCount(info.getDisconnectedCount() + 1);
                        }*/
                        info.setDisconnectedCount(info.getDisconnectedCount() + 1);
                        if (info.getDisconnectedCount() == 11) {
                            info.setWifiStatus(DeviceInfo.Status.DisConnected);
                            mDeviceManger.update(info);
                        }
                        LogUtil.eee(info.getHostId() + " disconnected count====" + info.getDisconnectedCount());
                    }
                }
            }, 1500, 1500);

            while (waitFlag && isConnected()) {
                try {
                    UdpPacket inPacket = receiveMessage(UdpSocket.PACKET_BUFFER_SIZE);
                    if (inPacket.hasData()) {
                        LogUtil.eee("beatworker receive Data from Remote: "
                                + inPacket.getSocketAddress());
                        int type = SocketUtils.byte2Int(inPacket.getMessage().getAction());
                        switch (type) {
                            case KEY_ACTION_GROUP_CONNECTED_REQUEST:
                                sendPacketData(new byte[0], new byte[]{KEY_ACTION_GROUP_CONNECTED_RESPONSE}, inPacket.getSocketAddress());
                                DeviceInfo deviceInfo = DeviceInfo.parse(new String(inPacket.getMessage().getPacketData()));
                                if (deviceInfo == null) {
                                    LogUtil.e("接收到的数据为空");
                                    break;
                                }
                                DeviceInfo deviceInfo1 = findDeviceInfo(deviceInfo.getHostId());
                                if (deviceInfo1 != null) {
                                    deviceInfo1.setDisconnectedCount(0);
                                }
                                break;
                            default:
                                //PromptUtil.poolPlay(mContext, R.raw.group_content_error);
                                break;
                        }
                    } else {
                        Log.i(TAG, "no data received");
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        @Override
        public void close() {
            super.close();
            if (timer != null) timer.cancel();
            System.gc();
        }
    }

    /**
     * 用于维护客户端数据传输
     */
    class VoiceResWorker extends NetWorker {
        private String inputAddress = null;

        public VoiceResWorker(UdpSocket udpSocket) {
            super(udpSocket);
        }

        @Override
        public void work() {
            while (waitFlag && isConnected()) {
                try {
                    UdpPacket inPacket = receiveMessage(UdpSocket.PACKET_BUFFER_SIZE);
                    try {
                        if (inPacket.hasData()) {
                            LogUtil.i(TAG, "VoiceResWorker ========= receive Data from Remote: "
                                    + inPacket.getSocketAddress());
                            try {
                                String tempAddress = inPacket.getSocketAddress().toString();
                                int type = SocketUtils.byte2Int(inPacket.getMessage().getAction());
                                switch (type) {
                                    case KEY_ACTION_VOICE_REQUEST_START:
                                        if (TextUtils.isEmpty(inputAddress)) {
                                            LogUtil.e(TAG, "VoiceResWorker ========= response permit");
                                            sendPacketData(inPacket.getMessage().getPacketData(), inPacket.getSocketAddress(), 1);
                                            inputAddress = tempAddress;
                                            PromptUtil.poolPlay(mContext, R.raw.prompt_permit, PromptUtil.TYPE_PLAY_POOL_INSTANT);
                                        } else {
                                            LogUtil.e(TAG, "VoiceResWorker ========= response prevent");
                                            sendPacketData(inPacket.getMessage().getPacketData(), inPacket.getSocketAddress(), 2);
                                        }
                                        break;
                                    case KEY_ACTION_VOICE_REQUEST_STOP:
                                        LogUtil.e(TAG, "VoiceResWorker ========= response finish");
                                        sendPacketData(inPacket.getMessage().getPacketData(), inPacket.getSocketAddress(), 3);
                                        if (TextUtils.equals(inputAddress, tempAddress)) {
                                            inputAddress = null;
                                        }
                                        break;
                                    case KEY_ACTION_VOICE_REQUEST_PERSIST:
                                        if (TextUtils.equals(inputAddress, tempAddress)) {
                                            LogUtil.e(TAG, "VoiceResWorker ========= response persist");
                                            sendPacketData(inPacket.getMessage().getPacketData(), inPacket.getSocketAddress(), 4);
                                        } else {
                                            LogUtil.e(TAG, "VoiceResWorker ========= response prevent");
                                            sendPacketData(inPacket.getMessage().getPacketData(), inPacket.getSocketAddress(), 2);
                                        }
                                        break;
                                    default:
                                        //PromptUtil.poolPlay(mContext, R.raw.group_content_error);
                                        break;
                                }
                            } catch (NullPointerException e) {
                                e.printStackTrace();
                            }
                        } else {
                            LogUtil.i(TAG, "no data received");
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        inputAddress = null;
                    }
                } catch (SocketTimeoutException e) {
                    LogUtil.eee("no voice input request or input is stopped or input is disconnected");
                    inputAddress = null;
                } catch (IOException e) {
                    e.printStackTrace();
                    inputAddress = null;
                }
            }
        }

        public void sendPacketData(byte[] data, SocketAddress beatAddr, int action) throws IOException {
            if (isConnected()) {
                byte[] actionBytes = new byte[]{0};
                if (action == 1) {
                    actionBytes = new byte[]{KEY_ACTION_VOICE_RESPONSE_PERMIT};
                } else if (action == 2) {
                    actionBytes = new byte[]{KEY_ACTION_VOICE_RESPONSE_PREVENT};
                } else if (action == 3) {
                    actionBytes = new byte[]{KEY_ACTION_VOICE_RESPONSE_FINISH};
                } else if (action == 4) {
                    actionBytes = new byte[]{KEY_ACTION_VOICE_RESPONSE_PERSIST};
                }
                sendPacketData(data, actionBytes, beatAddr);
            } else {
                LogUtil.eee("UDP Socket Status is " + " , Current Net Status: ");
            }
        }
    }
}
